#ifndef MESSAGE_H
#define MESSAGE_H
#include "../common/message.pb.h"
#include "../common/Helper.hpp"
#include <cassert>
#include <cstddef>
#include <list>
#include <memory>
#include <mutex>
#include <string>
#include <unordered_map>
using namespace std;

#define DATAFILE_SUBFIX ".mqd"
#define TMPFILE_SUBFIX ".mqd.tmp"

namespace mymessageptr {
    using MessagePtr = shared_ptr<MessageQueue::Message>;
}

// 使用队列来管理消息，每个队列有自己的文件
// 消息在磁盘中的管理
class MessageMapper
{
public:
    MessageMapper(string& basedir, const string& qname) :
        _qname(qname)
    {
        if(basedir.back() != '/') basedir.push_back('/');
        // 构造出队列的文件信息
        _datafile = basedir + qname + DATAFILE_SUBFIX;
        _tmpfile = basedir + qname + TMPFILE_SUBFIX;
        // 如果目录不存在就创建
        if(FileHelper(basedir).exists() == false)
            assert(FileHelper::createDirectory(basedir));
        CreateMessageFile();
    }
    
    bool CreateMessageFile()
    {
        if(FileHelper(_datafile).exists() == true)
            return true;
        bool ret = FileHelper::createFile(_datafile);
        if(ret == false)
        {
            lg(LOG_ERROR_CODE, "创建队列数据文件 %s 失败", _datafile.c_str());
            return false;
        }
        return true;
    }

    void RemoveMessageFile()
    {
        FileHelper::removeFile(_datafile);
        FileHelper::removeFile(_tmpfile);
    }

    bool Insert(mymessageptr::MessagePtr& msg)
    {
        return Insert(_datafile, msg);      
    }

    bool Remove(mymessageptr::MessagePtr& msg)
    {
        // 1. 修改标记位
        msg->mutable_payload()->set_valid("0");
        // 2. 对msg进行序列化
        string body = msg->payload().SerializeAsString();
        if(body.size() != msg->length())
        {
            lg(Warning, "不能修改数据信息，长度不一致");
            return false;
        }
        // 3. 把序列化信息写到数据在文件的指定位置，直接覆盖
        FileHelper helper(_datafile);
        bool ret = helper.write(body.c_str(), msg->offset(), body.size());
        if(ret == false)
        {
            lg(LOG_ERROR_CODE, "写入数据失败");
            return false;
        }
        return true;
    }

    // 垃圾回收
    list<mymessageptr::MessagePtr> gc()
    {
        // 1. 把有效信息塞到链表中
        list<mymessageptr::MessagePtr> result;
        bool ret = load(result);
        if(ret == false)
        {
            lg(LOG_ERROR_CODE, "加载有效数据失败");
            return result;
        }
        lg(Debug, "垃圾回收，有效信息数量为: %d", result.size());
        // 2. 将有效数据进行序列化到临时文件
        FileHelper::createFile(_tmpfile);
        for(auto& msg : result)
        {
            lg(Debug, "向临时文件中写数据: %s", msg->payload().body().c_str());
            ret = Insert(_tmpfile, msg);
            if(ret == false)
            {
                lg(LOG_ERROR_CODE, "向临时文件写数据失败");
                return result;
            }
        }
        lg(Debug, "垃圾回收后，向临时文件写数据后，临时文件大小: %ld", FileHelper(_tmpfile).size());
        // 3. 删除源文件
        ret = FileHelper::removeFile(_datafile);
        if(ret == false)
        {
            lg(LOG_ERROR_CODE, "删除文件失败: %s", _datafile.c_str());
            return result;
        }
        // 4. 修改临时文件名为源文件名
        ret = FileHelper(_tmpfile).rename(_datafile);
        if(ret == false)
        {
            lg(LOG_ERROR_CODE, "重命名失败: %s", _datafile.c_str());
            return result;
        }
        // 5. 返回有效数据
        return result;
    }

private:
    // 把文件信息写到链表中
    bool load(list<mymessageptr::MessagePtr>& result)
    {
        // 1. 加载文件的有效数据
        FileHelper data_file_helper(_datafile);
        size_t offset = 0, msg_size = 0;
        size_t fsize = data_file_helper.size();
        bool ret;
        while(offset < fsize)
        {
            ret = data_file_helper.read((char*)&msg_size, offset, sizeof(size_t));
            if(ret == false)
            {
                lg(LOG_ERROR_CODE, "读取消息长度失败");
                return false;
            }
            offset += sizeof(size_t);
            string msg_body(msg_size, '\0');
            data_file_helper.read(&msg_body[0], offset, msg_size);
            if(ret == false)
            {
                lg(LOG_ERROR_CODE, "读取信息失败");
                return false;
            }
            offset += msg_size;
            mymessageptr::MessagePtr msgp = make_shared<MessageQueue::Message>();
            msgp->mutable_payload()->ParseFromString(msg_body);
            // 如果是无效信息，就加载下一个
            if(msgp->payload().valid() == "0")
            {
                lg(Debug, "加载到无效信息: %s", msgp->payload().body().c_str());
                continue;
            }
            lg(Debug, "加载到有效数据: %s", msgp->payload().body().c_str());
            result.push_back(msgp);
        }
        return true;
    }

    // 向指定的文件中插入某个特定信息
    bool Insert(const string& filename, mymessageptr::MessagePtr& msg)
    {
        // 1. 把消息进行序列化
        string body = msg->payload().SerializeAsString();
        // 2. 获取文件长度
        FileHelper helper(filename);
        size_t fsize = helper.size();
        size_t msg_size = body.size();
        // 3. 进行依次写入，数据长度+数据
        bool ret = helper.write((char*)&msg_size, fsize, sizeof(size_t));
        if(ret == false)
        {
            lg(LOG_ERROR_CODE, "向队列%s写入数据长度失败", filename.c_str());
            return false;
        }
        ret = helper.write(body.c_str(), fsize + sizeof(size_t), body.size());
        if(ret == false)
        {
            lg(LOG_ERROR_CODE, "向队列%s写入数据失败", filename.c_str());
            return false;
        }
        // 4. 更新msg中实际存储信息
        msg->set_offset(fsize + sizeof(size_t));
        msg->set_length(body.size());
        return true;
    }

private:
    string _qname;
    string _datafile;
    string _tmpfile;
};

// 消息在内存中的管理
class QueueMessage
{
public:
    using ptr = std::shared_ptr<QueueMessage>;
    QueueMessage(string& basedir, const string& qname)
        : _mapper(basedir, qname), _qname(qname), _valid_count(0), _total_count(0)
    {}

    // 恢复历史消息
    bool Recovery()
    {
        unique_lock<mutex> lock(_mutex);
        _msgs = _mapper.gc();
        for(auto& msg : _msgs)
            _durable_msgs.insert(make_pair(msg->payload().properties().id(),msg));
        _valid_count = _total_count = _msgs.size();
        return true;
    }

    // 插入消息
    bool Insert(const MessageQueue::BasicProperties *bp, const string& body, bool queue_is_durable)
    {
        // 1. 先把消息对象构造出来
        mymessageptr::MessagePtr msg = make_shared<MessageQueue::Message>();
        msg->mutable_payload()->set_body(body);
        // 根据特性构造内容
        if(bp != nullptr)
        {
            MessageQueue::DeliveryMode mode = queue_is_durable ? bp->delivery_mode() : MessageQueue::DeliveryMode::UNDURABLE;
            msg->mutable_payload()->mutable_properties()->set_id(bp->id());
            msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
            msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
        }
        // 如果没有特性，就默认带一点即可
        else
        {
            MessageQueue::DeliveryMode mode = queue_is_durable ? MessageQueue::DeliveryMode::DURABLE : MessageQueue::DeliveryMode::UNDURABLE;
            msg->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());
            msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
            msg->mutable_payload()->mutable_properties()->set_routing_key("");
        }
        // 2. 把消息进行持久化存储
        if(msg->payload().properties().delivery_mode() == MessageQueue::DeliveryMode::DURABLE)
        {
            // 表示持久化存储数据有效
            msg->mutable_payload()->set_valid("1");
            bool ret = _mapper.Insert(msg);
            if(ret == false)
            {
                lg(LOG_ERROR_CODE, "持久化存储消息：%s 失败了", body.c_str());
                return false;
            }
            _valid_count++;
            _total_count++;
            _durable_msgs.insert(make_pair(msg->payload().properties().id(), msg));
        }
        // 3. 进行内存管理
        _msgs.push_back(msg);
        return true;
    }

    // 获取队首数据
    mymessageptr::MessagePtr Front()
    {
        unique_lock<mutex> lock(_mutex);
        if(_msgs.size() == 0)
            return mymessageptr::MessagePtr();
        // 从_msgs中把消息取出来
        mymessageptr::MessagePtr msg = _msgs.front();
        _msgs.pop_front();
        // 把消息对象放到待确认hash表中
        _wait_msgs.insert(make_pair(msg->payload().properties().id(), msg));
        return msg;
    }

    // 每次删除消息后，判断是否需要垃圾回收
    bool Remove(const string& msg_id)
    {
        unique_lock<mutex> lock(_mutex);
        // 1. 从待确认队列中找消息
        auto it = _wait_msgs.find(msg_id);
        if(it == _wait_msgs.end())
        {
            lg(Warning, "没有找到要删除的消息：%s", msg_id.c_str());
            return true;
        }
        // 2. 删除持久化消息
        if(it->second->payload().properties().delivery_mode() == MessageQueue::DeliveryMode::DURABLE)
        {
            _mapper.Remove(it->second);
            _durable_msgs.erase(msg_id);
            _valid_count--;
            // 判断是否要进行垃圾回收
            gc();
        }
        // 4. 删除内存中的消息
        _wait_msgs.erase(msg_id);
        return true;
    }

    size_t GetableCount()
    {
        unique_lock<std::mutex> lock(_mutex);
        return _msgs.size();
    }
    size_t TotalCount()
    {
        unique_lock<std::mutex> lock(_mutex);
        return _total_count;
    }
    size_t DurableCount()
    {
        unique_lock<std::mutex> lock(_mutex);
        return _durable_msgs.size();
    }
    size_t WaitAckCount()
    {
        unique_lock<std::mutex> lock(_mutex);
        return _wait_msgs.size();
    }
    void clear()
    {
        unique_lock<std::mutex> lock(_mutex);
        _mapper.RemoveMessageFile();
        _msgs.clear();
        _durable_msgs.clear();
        _wait_msgs.clear();
        _valid_count = 0;
        _total_count = 0;
    }

private:
    bool GCCheck()
    {
        // 如果持久化消息数量满足策略就进行垃圾删除
        if(_total_count > 2000 && _valid_count * 10 / _total_count < 5)
            return true;
        return false;
    }

    void gc()
    {
        // 1. 进行垃圾回收，获取有效的消息链表
        if(GCCheck() == false)
            return;
        list<mymessageptr::MessagePtr> msgs = _mapper.gc();
        for(auto& msg : msgs)
        {
            auto it = _durable_msgs.find(msg->payload().properties().id());
            if(it == _durable_msgs.end())
            {
                lg(Warning, "垃圾回收后有持久化消息没有在内存中: %s", msg->payload().properties().id().c_str());
                // 把它加进去就行了
                _msgs.push_back(msg);
                _durable_msgs.insert(make_pair(msg->payload().properties().id(), msg));
                continue;
            }
            // 2. 更新实际存储位置
            it->second->set_offset(msg->offset());
            it->second->set_length(msg->length());
        }
        // 3. 更新当前有效信息数量
        _valid_count = _total_count = msgs.size();
    }
private:
    string _qname;
    size_t _valid_count;
    size_t _total_count;
    MessageMapper _mapper;
    list<mymessageptr::MessagePtr> _msgs;
    unordered_map<string, mymessageptr::MessagePtr> _durable_msgs;
    unordered_map<string, mymessageptr::MessagePtr> _wait_msgs;
    mutex _mutex;
};

// 对外提供操控消息的总接口
class MessageManager
{
public:
    using ptr = shared_ptr<MessageManager>;
    MessageManager(const string& basedir) : _basedir(basedir)
    {}

    void clear()
    {
        unique_lock<mutex> lock(_mutex);
        // 清空每一个队列的消息内容
        for(auto& qmsg : _queue_msgs)
            qmsg.second->clear();
    }

    // 把队列进行初始化获取消息
    void InitQueueMessage(const string& qname)
    {
        QueueMessage::ptr qmp;
        {
            unique_lock<mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if(it != _queue_msgs.end())
                return;
            qmp = make_shared<QueueMessage>(_basedir, qname);
            _queue_msgs.insert(make_pair(qname, qmp));
        }
        qmp->Recovery();
    }

    // 销毁队列
    void DestroyQueueMessage(const string& qname)
    {
        QueueMessage::ptr qmp;
        {
            unique_lock<mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if(it == _queue_msgs.end())
                return;
            qmp = it->second;
            _queue_msgs.erase(it);
        }
        qmp->clear();
    }

    // 插入消息
    bool Insert(const string& qname, MessageQueue::BasicProperties* bp, const string& body, bool queue_is_durable)
    {
        QueueMessage::ptr qmp;
        {
            unique_lock<mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if(it == _queue_msgs.end())
            {
                lg(Warning, "向队列：%s 新增消息失败：没有找到消息管理句柄", qname.c_str());
                return false;
            }
            qmp = it->second;
        }
        return qmp->Insert(bp, body, queue_is_durable);
    }

    // 获取队首消息信息
    mymessageptr::MessagePtr Front(const string& qname)
    {
        QueueMessage::ptr qmp;
        {
            unique_lock<mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if(it == _queue_msgs.end())
            {
                lg(Warning, "向队列：%s 获取队首消息失败：没有找到消息管理句柄", qname.c_str());
                return mymessageptr::MessagePtr();
            }
            qmp = it->second;
        }
        return qmp->Front();
    }

    // 获取ack应答
    void ack(const string& qname, const string& msg_id)
    {
        QueueMessage::ptr qmp;
        {
            unique_lock<mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if(it == _queue_msgs.end())
            {
                lg(Warning, "确认队列%s消息%s失败：没有找到消息管理句柄!", qname.c_str(), msg_id.c_str());
                return;
            }
            qmp = it->second;
        }
        qmp->Remove(msg_id);
        return;
    }

    size_t GetableCount(const string& qname)
    {
        QueueMessage::ptr qmp;
        {
            unique_lock<mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if(it == _queue_msgs.end())
            {
                lg(Warning, "获取队列%s待推送消息数量失败：没有找到消息管理句柄!", qname.c_str());
                return 0;
            }
            qmp = it->second;
        }
        return qmp->GetableCount();
    }

    size_t TotalCount(const string& qname)
    {
        QueueMessage::ptr qmp;
        {
            unique_lock<mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if(it == _queue_msgs.end())
            {
                lg(Warning, "获取队列%s总持久化消息数量失败：没有找到消息管理句柄!", qname.c_str());
                return 0;
            }
            qmp = it->second;
        }
        return qmp->TotalCount();
    }

    size_t DurableCount(const string& qname)
    {
        QueueMessage::ptr qmp;
        {
            unique_lock<mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if(it == _queue_msgs.end())
            {
                lg(Warning, "获取队列%s有效持久化消息数量失败：没有找到消息管理句柄!", qname.c_str());
                return 0;
            }
            qmp = it->second;
        }
        return qmp->DurableCount();
    }

    size_t WaitAckCount(const string& qname)
    {
        QueueMessage::ptr qmp;
        {
            unique_lock<mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if(it == _queue_msgs.end())
            {
                lg(Warning, "获取队列%s待确认消息数量失败：没有找到消息管理句柄!", qname.c_str());
                return 0;
            }
            qmp = it->second;
        }
        return qmp->WaitAckCount();
    }
private:
    mutex _mutex;
    string _basedir;
    unordered_map<string, QueueMessage::ptr> _queue_msgs;
};

#endif